#include "wordcount_mapred.h"
#include <ctype.h>
#include <assert.h>

enum {
   IN_WORD = 0,
   NOT_IN_WORD
};

int mystrcmp(const void *s1, const void *s2)
{
   return strcmp((const char *)s1, (const char *) s2);
}


int mykeyvalcmp(const void *v1, const void *v2)
{
   keyval_t* kv1 = (keyval_t*)v1;
   keyval_t* kv2 = (keyval_t*)v2;

   int i1 = (int)kv1->val;
   int i2 = (int)kv2->val;

   if (i1 < i2) return 1;
   else if (i1 > i2) return -1;
   else {
      /*** don't just return 0 immediately coz the mapreduce scheduler provides 
   1 key with multiple values to the reduce task. since the different words that 
   we require are each part of a different keyval pair, returning 0 makes
   the mapreduce scheduler think that it can just keep one key and disregard
   the rest. That's not desirable in this case. Returning 0 when the values are
   equal produces results where the same word is repeated for all the instances
   which share the same frequency. Instead, we check the word as well, and only 
   return 0 if both the value and the word match ****/
      return strcmp((char *)kv1->key, (char *)kv2->key);
      //return 0;
   }
}



int wordcount_splitter(void *data_in, int req_units, map_args_t *out)
{
   wc_data_t * data = (wc_data_t *)data_in; 
   
   assert(data_in);
   assert(out);
   
   assert(data->flen >= 0);
   assert(data->fdata);
   assert(req_units);
   assert(data->fpos >= 0);

   // End of file reached, return FALSE for no more data
   if (data->fpos >= data->flen) return 0;

   // Set the start of the next data
   out->data = (void *)&data->fdata[data->fpos];
   
   // Determine the nominal length
   out->length = req_units * data->unit_size;
   
   
   if (data->fpos + out->length > data->flen)
      out->length = data->flen - data->fpos;
   
   // Set the length to end at a space
   for (data->fpos += (long)out->length;
        data->fpos < data->flen && 
        data->fdata[data->fpos] != ' ' && data->fdata[data->fpos] != '\t' &&
        data->fdata[data->fpos] != '\r' && data->fdata[data->fpos] != '\n';
        data->fpos++, out->length++);
  
   return 1;
}


void wordcount_map(map_args_t *args) 
{
   char *curr_start, curr_ltr;
   int state = NOT_IN_WORD;
   int i;
  
   assert(args);

   char *data = (char *)args->data;

   assert(data);
   curr_start = data;
   
   for (i = 0; i < args->length; i++)
   {
     //curr_ltr = toupper(data[i]);
      curr_ltr = tolower(data[i]);
      switch (state)
      {
      case IN_WORD:
         data[i] = curr_ltr;
         if ((curr_ltr < 'a' || curr_ltr > 'z') && curr_ltr != '\'')
         {
            data[i] = 0;
            emit_intermediate(curr_start, (void *)1, &data[i] - curr_start + 1);
            state = NOT_IN_WORD;
         }
      break;

      default:
      case NOT_IN_WORD:
         if (curr_ltr >= 'a' && curr_ltr <= 'z')
         {
            curr_start = &data[i];
            data[i] = curr_ltr;
            state = IN_WORD;
         }
         break;
      }
   }

   // Add the last word
   if (state == IN_WORD)
   {
      data[args->length] = 0;
      emit_intermediate(curr_start, (void *)1, &data[i] - curr_start + 1);
   }
}



void wordcount_reduce(void *key_in, void **vals_in, int vals_len) 
{
   char * key = (char *)key_in;
   int  * vals = (int*)vals_in;
   int i, sum = 0;

   assert(key);
   assert(vals);

   for (i = 0; i < vals_len; i++)
      sum += vals[i];

   emit(key, (void *)sum);
}
